package mqtt

import (
	"crypto/tls"
	"errors"
	"gitee.com/injoyai/goutil/chans"
	"gitee.com/injoyai/goutil/conv"
	"gitee.com/injoyai/goutil/string/logs"
	mqtt "github.com/eclipse/paho.mqtt.golang"
)

type Client struct {
	mqtt.Client
	config      *Config
	chanMessage chan []byte
	err         *chans.Err
}

func (this *Client) Cfg() *Config {
	return this.config
}

func (this *Client) Err() error {
	return this.err.Err()
}

func (this *Client) Debug(b ...bool) {
	this.config.Debug = !(len(b) > 0 && !b[0])
}

// ReadMessage 读取一条数据,先进先出,实现io.MessageReader接口
func (this *Client) ReadMessage() ([]byte, error) {
	return <-this.chanMessage, nil
}

// Close 关闭链接,实现io.Closer接口
func (this *Client) Close() error {
	this.Client.Disconnect(0)
	return this.err.Close()
}

// Publish 发布数据,如果是阿里云则需要加上前缀,自行拼接 "xxx/p2p/xxx@@@"+"xxx"
func (this *Client) Publish(target string, msg []byte) (err error) {
	//打印数据,保存记录
	defer func() {
		this.err.CloseWithErr(err)
		if this.config.Debug {
			logs.Debug("[发送]", string(msg), " 到 ", target, conv.New(err).String("成功"))
		}
	}()
	token := this.Client.Publish(target, this.config.Qos, this.config.Retained, msg)
	if !token.WaitTimeout(this.config.Timeout) {
		return errors.New("发送超时")
	}
	if token.Error() != nil {
		return token.Error()
	}
	return
}

// handler 同步数据处理
func (this *Client) handler(c mqtt.Client, msg mqtt.Message) {
	this.chanMessage <- msg.Payload()
}

func New(cfg *Config) (client *Client, err error) {
	client = &Client{
		config:      cfg,
		chanMessage: make(chan []byte),
		err:         chans.NewErr(),
	}
	var tls *tls.Config
	if cfg.TLS {
		tls, err = cfg.TLSConfig.New()
		if err != nil {
			return nil, err
		}
	}
	client.Client = mqtt.NewClient(
		mqtt.NewClientOptions().
			SetTLSConfig(tls).
			AddBroker(cfg.BrokerURL).
			SetClientID(cfg.ClientID).
			SetUsername(cfg.Username).
			SetPassword(cfg.Password).
			SetDefaultPublishHandler(client.handler).
			SetConnectTimeout(cfg.Timeout).
			SetKeepAlive(cfg.KeepAlive).
			SetAutoReconnect(true). //自动重连
			SetCleanSession(false)) //重连后恢复session
	token := client.Client.Connect()
	token.Wait()
	logs.Infof("MQTT(%s)连接 %s\n", cfg.ClientID, conv.New(token.Error()).String("成功"))
	return client, token.Error()
}
